package jupitermouse.site.sink.custom.mysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import org.apache.flink.api.java.tuple.Tuple2;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @desc 自定义数据源写入mysql
 * @author renqiqiang
 */
@Slf4j
public class MysqlSinkFunction extends RichSinkFunction<Tuple2<String,Long>> {


    /**
     * 预执行
     */
    private PreparedStatement ps;

    /**
     * 连接
     */
    private Connection connection;

    @Override
    public void open(Configuration conf) throws Exception {
        super.open(conf);
        log.info("start connection...");
        String driverClass = conf.getString("custom.rdbms.driverClass", "com.mysql.jdbc.Driver");
        String url = conf.getString("custom.rdbms.url", "jdbc:mysql://127.0.0.1:3306/flink?useSSL=false");
        String username = conf.getString("custom.rdbms.username", "root");
        String password = conf.getString("custom.rdbms.password", "root");

        connection = getConnection(driverClass, url, username, password);
        ps  = connection.prepareStatement("insert into wordcount(word,count) values(?,?); ");
        log.info("Thread:{},connection end ...", Thread.currentThread().getName());
    }

    @Override
    public void invoke(Tuple2<String, Long> value, Context context) throws Exception {
        ps.setString(1, value.f0);
        ps.setLong(2, value.f1);
        log.info("sql:insert into wordcount(word,count) values({},{}); ",value.f0, value.f1);
        ps.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭连接和释放资源
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }


    /**
     * 获取数据源连接
     * @param driverClass 驱动类地址
     * @param url url
     * @param username 用户名
     * @param password 密码
     * @return Connection
     */
    private static Connection getConnection(String driverClass, String url, String username, String password){
        Connection connection = null;
        try {
            Class.forName(driverClass);
        } catch (Exception e) {
            log.info("driverClass:{} error", driverClass);
        }
        try {
            connection = DriverManager.getConnection(url, username, password);
        } catch (SQLException e) {
            log.info("connection: error \n url:{}\n username:{}\n password:{}\n", url, username, password);
        }
        return connection;
    }
}
